package com.atguigu.edu.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import util.MyKafkaUtil;

import java.time.Duration;

/*
        用户表：下单事实表
 */
public class DwdInteractionOrderPerProcess {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //设置状态的TTL
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(16*60+5));
        //检查点相关设置  略
        //从kafka的ODS_BASE_DB主题中读取数据  创建 动态表
        tableEnv.executeSql(MyKafkaUtil.getTopicDDL("dwd_interaction_order_per"));
        //筛选订单明细数据
        Table orderDetail = tableEnv.sqlQuery("select \n" +
                "data['id'] id,\n" +
                "data['course_id'] course_id,\n" +
                "data['course_name'] course_name,\n" +
                "data['order_id'] order_id,\n" +
                "data['user_id'] user_id,\n" +
                "data['origin_amount'] origin_amount,\n" +
                "data['coupon_reduce'] coupon_reduce,\n" +
                "data['final_amount'] final_amount,\n" +
                "data['session_id'] session_id,\n" +
                "data['create_time'] create_time,\n" +
                "data['update_time'] update_time,\n" +
                "ts od_ts,\n" +
                "proc_time\n" +
                "from `ODS_BASE_DB` where `table` = 'order_detail' " +
                "and `type` = 'insert'\n");
        tableEnv.createTemporaryView("orderDetail",orderDetail);
        //  筛选订单数据


        Table orderInfo = tableEnv.sqlQuery("select \n" +
                "data['id'] id,\n" +
                "data['user_id'] user_id,\n" +
                "data['origin_amount'] origin_amount,\n" +
                "data['coupon_reduce'] coupon_reduce,\n" +
                "data['final_amount'] final_amount,\n" +
                "data['order_status'] order_status,\n" +
                "data['out_trade_no'] out_trade_no,\n" +
                "data['trade_body'] trade_body,\n" +
                "data['session_id'] session_id,\n" +
                "data['province_id'] province_id,\n" +
                "data['create_time'] create_time,\n" +
                "data['expire_time'] expire_time,\n" +
                "data['update_time'] update_time,\n" +
                "`type`,\n" +
                "`old`,\n" +
                "ts oi_ts\n" +
                "from `ODS_BASE_DB`\n" +
                "where `table` = 'order_info'\n" +
                "and (`type` = 'insert' or `type` = 'update')");
        tableEnv.createTemporaryView("order_info",orderInfo);

        //进行关联
        Table resultTable = tableEnv.sqlQuery("select\n" +
                "od.id\n" +
                ",od.course_id\n" +
                ",od.course_name\n" +
                ",od.order_id\n" +
                ",od.user_id\n" +
                ",od.origin_amount\n" +
                ",od.final_amount\n" +
                ",od.session_id\n" +
                ",od.od_ts od_ts\n" +

                ",oi.order_status\n" +
                ",oi.out_trade_no\n" +
                ",oi.province_id\n" +
                ",oi.create_time\n" +
                ",oi.expire_time\n" +
                ",oi.update_time\n" +
                ",oi.type\n" +
                ",oi.`old`\n" +
                ",oi.oi_ts oi_ts\n" +
                ",od.proc_time\n" +
                "from orderDetail od \n" +
                "join order_info oi\n" +
                "on od.order_id=oi.id");
        tableEnv.createTemporaryView("result_table",resultTable);
      //  tableEnv.executeSql("select * from " + resultTable).print();

    //创建动态表和要写入的kafka主题进行映射
        tableEnv.executeSql("create table dwd_interaction_order_per(\n" +
                "id string,\n" +
                "course_id string,\n" +
                "course_name string,\n" +
                "order_id string,\n" +
                "user_id string,\n" +
                "origin_amount string,\n" +
                "final_amount string,\n" +
                "session_id string,\n" +
                "od_ts string,\n" +
                "order_status string,\n" +
                "out_trade_no string,\n" +
                "province_id string,\n" +
                "create_time string,\n" +
                "expire_time string,\n" +
                "update_time string,\n" +
                "`type` string,\n" +
                "`old` map<string,string>,\n" +
                "oi_ts string,\n" +
                "row_op_ts timestamp_ltz(3),\n" +
                "primary key(id) not enforced\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_interaction_order_per"));

        //将关联的结果写到kafka主题中
        tableEnv.executeSql("insert into dwd_interaction_order_per \n" +
                "select * from result_table");
    }
}
